package org.rrd.kafka.phoneix.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @program: MysqlCKFKDruid
 * @description:
 * @author: sunteng
 * @create: 2019-08-23 14:43
 **/
@Slf4j
public class FlinkUse {

    public static void setEnv(StreamExecutionEnvironment env) {
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    }

    public static SingleOutputStreamOperator<String> setWaterMark(DataStream<String> stream) {
        SingleOutputStreamOperator<String> stringSingleOutputStreamOperator = stream.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(String element) {
                        long result = 0;
                        try {
                            log.info("the element message is ..........{}", element);
                            JSONObject record = JSON.parseObject(element);

                            result = record.getLong("ts");
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        return result;
                    }

                }
        );
        return stringSingleOutputStreamOperator;
    }

}
